package com.example.sql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/**
 * Author wangJinLong
 * Date 2025/8/22 16:27
 **/
public class FlinkSqlWindow {
    public static void main(String[] args) {
        groupBy();

    }

    private static void topN() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateBid(environment);

        // ROW 间隔
        /*environment.executeSql(
                "SELECT item, bidtime, price,\n" +
                        "  SUM(price) OVER w AS sum_price,\n" +
                        "  AVG(price) OVER w AS avg_price\n" +
                        "FROM Bid2\n" +
                        "WINDOW w AS (\n" +
                        "  PARTITION BY supplier_id\n" +
                        "  ORDER BY bidtime\n" +
                        "  RANGE BETWEEN INTERVAL '30' SECONDS PRECEDING AND CURRENT ROW\n" +
                        ")"
        ).print();*/

        // Top N
        /*environment.executeSql(
                "SELECT *
                FROM (
                    SELECT *,
                    ROW_NUMBER() OVER (PARTITION BY supplier_id ORDER BY bidtime DESC) AS row_num
                    FROM Bid2
                )
                WHERE row_num <= 5"
        ).print();*/

        // 窗口top-n
        /*environment.executeSql(
                "SELECT *
                FROM (
                    SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
                    FROM (
                        SELECT window_start, window_end, supplier_id, SUM(price) as price, COUNT(*) as cnt
                        FROM TABLE(
                            TUMBLE(TABLE Bid2, DESCRIPTOR(bidtime), INTERVAL '10' SECONDS)
                        )
                        GROUP BY window_start, window_end, supplier_id
                    )
                ) WHERE rownum <= 3;"
        ).print();*/

        // 去重
        environment.executeSql(
                "SELECT supplier_id, item, price, bidtime\n" +
                        "FROM (\n" +
                        "  SELECT *,\n" +
                        "    ROW_NUMBER() OVER (PARTITION BY supplier_id ORDER BY bidtime desc) AS row_num\n" +
                        "  FROM Bid2\n" +
                        ")\n" +
                        "WHERE row_num = 1;"
        ).print();
    }

    private static void groupBy() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateBid(environment);

        // 分组集
        /*environment.executeSql("SELECT supplier_id, rating, COUNT(*) AS total\n" +
                "FROM (VALUES\n" +
                "    ('supplier1', 'product1', 4),\n" +
                "    ('supplier1', 'product2', 3),\n" +
                "    ('supplier2', 'product3', 3),\n" +
                "    ('supplier2', 'product4', 4))\n" +
                "AS Products(supplier_id, product_id, rating)\n" +
                "GROUP BY GROUPING SETS ((supplier_id, rating), (supplier_id), ());"
        ).print();*/

        // 汇总
        /*environment.executeSql(
                "SELECT supplier_id, rating, COUNT(*)\n" +
                        "FROM (VALUES\n" +
                        "    ('supplier1', 'product1', 4),\n" +
                        "    ('supplier1', 'product2', 3),\n" +
                        "    ('supplier2', 'product3', 3),\n" +
                        "    ('supplier2', 'product4', 4))\n" +
                        "AS Products(supplier_id, product_id, rating)\n" +
                        "GROUP BY ROLLUP (supplier_id, rating)"
        ).print();*/

        // 立方体
        /*environment.executeSql(
                "SELECT supplier_id, rating, product_id, COUNT(*)\n" +
                        "FROM (VALUES\n" +
                        "    ('supplier1', 'product1', 4),\n" +
                        "    ('supplier1', 'product2', 3),\n" +
                        "    ('supplier2', 'product3', 3),\n" +
                        "    ('supplier2', 'product4', 4))\n" +
                        "AS Products(supplier_id, product_id, rating)\n" +
                        "GROUP BY CUBE (supplier_id, rating, product_id)"
        ).print();*/

        // OVER 聚合
        environment.executeSql(
                "SELECT item, bidtime, price,\n" +
                        "  SUM(price) OVER (\n" +
                        "    PARTITION BY supplier_id\n" +
                        "    ORDER BY bidtime\n" +
                        "    RANGE BETWEEN INTERVAL '10' SECONDS PRECEDING AND CURRENT ROW\n" +
                        "  ) AS price_sum\n" +
                        "FROM Bid2;"
        ).print();


    }

    private static void window() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateBid(environment);

        // 滚动窗口
        /*environment.executeSql(
                "SELECT window_start, window_end, SUM(price)\n" +
                "  FROM TABLE(\n" +
                "    TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' SECONDS))\n" +
                "  GROUP BY window_start, window_end;"
        ).print();*/

        // 滑动窗口
        /*environment.executeSql(
                "SELECT window_start, window_end, SUM(price)\n" +
                "  FROM TABLE(\n" +
                "    HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' SECONDS, INTERVAL '10' SECONDS))\n" +
                "  GROUP BY window_start, window_end;"
        ).print();*/

        // 累积窗口
        /*environment.executeSql(
                "SELECT window_start, window_end, SUM(price)\n" +
                " FROM TABLE(\n" +
                " CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' SECONDS, INTERVAL '10' SECONDS))\n" +
                " GROUP BY window_start, window_end;"
         ).print();*/

        // 分组集
        environment.executeSql(
                "SELECT window_start, window_end, item, SUM(price) as price\n" +
                "  FROM TABLE(\n" +
                "    TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' SECONDS))\n" +
                "  GROUP BY window_start, window_end, GROUPING SETS ((item), ());"
        ).print();
    }

    private static void generateBid(TableEnvironment environment) {
        environment.executeSql("CREATE TABLE Bid (\n" +
                "  item STRING,\n" +
                "  price DECIMAL(5, 2),\n" +
                "  bidtime TIMESTAMP_LTZ(3),\n" +
                "  WATERMARK FOR bidtime AS bidtime - INTERVAL '10' SECOND\n" +
                ")\n" +
                "WITH (\n" +
                "  'connector'='datagen',\n" +
                "  'rows-per-second' = '1',\n" +
                "  'fields.item.length' = '10'\n" +
                ");");

        environment.executeSql("CREATE TABLE Bid2 (\n" +
                "supplier_id STRING,\n" +
                "item STRING,\n" +
                "price DECIMAL(5, 2),\n" +
                "bidtime TIMESTAMP_LTZ(3),\n" +
                "WATERMARK FOR bidtime AS bidtime - INTERVAL '10' SECOND\n" +
                ")\n" +
                "WITH (\n" +
                "'connector'='datagen',\n" +
                "'rows-per-second' = '1',\n" +
                "'fields.item.length' = '2',\n" +
                "'fields.supplier_id.length' = '1'\n" +
                ");");
    }
}
